package org.ws.task.netty.support;

import com.google.common.collect.Maps;
import com.google.common.util.concurrent.Uninterruptibles;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.ChannelPromise;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultThreadFactory;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.ws.task.builder.RequestMessageBuilder;
import org.ws.task.event.support.startup.BootstrapEvent;
import org.ws.task.event.support.startup.BootstrapEventType;
import org.ws.task.event.support.BootstrapEventHolder;
import org.ws.task.common.AddressUtils;
import org.ws.task.exception.LeaderNotActiveException;
import org.ws.task.model.NodeMetadata;
import org.ws.task.netty.handler.ActionHandlerRegistry;
import org.ws.task.netty.NettyClient;
import org.ws.task.netty.NettyConfig;
import org.ws.task.netty.codec.MessageDecoder;
import org.ws.task.netty.codec.MessageEncoder;
import org.ws.task.model.protocol.base.RequestAction;
import org.ws.task.model.protocol.base.RequestMessage;
import org.ws.task.model.protocol.base.ResponseMessage;
import org.ws.task.repository.NodeMetadataRepository;

import java.net.SocketAddress;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;

public class DefaultNettyClient extends AbstractNettyService implements NettyClient {

    private static Logger log = LoggerFactory.getLogger(DefaultNettyClient.class.getName());

    /**
     * 所有Task的Leader都是一致
     */
    private volatile String leaderAddress;

    private final Bootstrap bootstrap;
    private final EventLoopGroup eventLoopGroup;
    private final CountDownLatch startCountDownLatch;

    private final ExecutorService callbackExecutor;

    private final Map<String, ChannelFuture> channelFutureMap = Maps.newConcurrentMap();
    private final Lock channelFutureMapLock = new ReentrantLock();
    private final AtomicBoolean running = new AtomicBoolean(false);
    private final AtomicBoolean active = new AtomicBoolean(false);
    private final AtomicInteger createChannelErrorCount = new AtomicInteger(0);
    private volatile int port = -1;

    public DefaultNettyClient(NettyConfig nettyConfig) {
        super(nettyConfig);
        this.startCountDownLatch = new CountDownLatch(1);
        this.bootstrap = new Bootstrap();
        // config
        this.eventLoopGroup = new NioEventLoopGroup(this.nettyConfig.getWorkerGroupThreadNum(),new DefaultThreadFactory("NettyClientSelector"));
        this.callbackExecutor = Executors.newFixedThreadPool(this.nettyConfig.getExecutorThreadNum(),new DefaultThreadFactory("NettyClientCallback"));
    }

    @Override
    protected ExecutorService getCallbackExecutor() {
        return this.callbackExecutor;
    }

    @Override
    public void start() {
        this.clearChannel();
        if(!this.running.get()){
            this.running.set(true);
            this.bootstrap.group(this.eventLoopGroup).channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    // config
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, this.nettyConfig.getConnectTimeoutMillis())
                    .option(ChannelOption.SO_SNDBUF, SO_SNDBUF)
                    .option(ChannelOption.SO_RCVBUF, SO_RCVBUF)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            pipeline.addLast(new LoggingHandler(LogLevel.DEBUG))
                                    .addLast(new IdleStateHandler(0,0,DefaultNettyClient.this.nettyConfig.getIdleTimeoutMillis()))
                                    .addLast(new MessageDecoder())
                                    .addLast(new MessageEncoder())
                                    .addLast(new ChannelConnectManageHandler())
                                    .addLast(new ResponseMessageHandler())
                                    .addLast(new RequestMessageHandler());

                        }
                    });
        }
        this.active.set(true);
        // 向leader发送一次心跳
        heartbeat();
        log.info("netty client start.");
    }

    @Override
    public int port() {
        return this.port;
    }

    @Override
    public boolean isActive() {
        return this.active.get();
    }

    @Override
    public void shutdown() {
        try {
            log.info("netty client shutdown");
            this.clearChannel();
            this.eventLoopGroup.shutdownGracefully();
            this.callbackExecutor.shutdown();
        } catch (Exception e) {
            log.error(e.getMessage(),e);
        }

        this.running.set(false);
    }

    @Override
    public void clearChannel(){
        for (ChannelFuture channelFuture : this.channelFutureMap.values()) {
            Channel channel = channelFuture.channel();
            try {
                closeChannel(channel);
            } catch (InterruptedException e) {
                log.error(e.getMessage(),e);
            }
        }
        this.channelFutureMap.clear();
    }

    @Override
    public boolean awaitStart(long timeout) {
        return Uninterruptibles.awaitUninterruptibly(this.startCountDownLatch,timeout, TimeUnit.MILLISECONDS);
    }

    private void closeChannel(Channel channel) throws InterruptedException {
        if(channel == null){
            return;
        }
        SocketAddress remoteAddress = channel.remoteAddress();
        final String address = AddressUtils.getStringSocketAddress(remoteAddress);
        if(this.channelFutureMapLock.tryLock(this.nettyConfig.getConnectTimeoutMillis(), TimeUnit.MILLISECONDS)){
            try {
                if(this.channelFutureMap.containsKey(address)) {
                    this.channelFutureMap.remove(address);
                    channel.close().addListener(future -> {
                        if(future.isSuccess()) {
                            log.info("close channel to address:{} success.", address);
                        }
                        else {
                            log.error("close channel to address:{} failed.", address);
                        }
                    });
                }
            }
            finally {
                this.channelFutureMapLock.unlock();
            }
        }
        else {
            // time out
        }
    }

    @Override
    public void close() throws Exception {
        shutdown();
    }

    @Override
    public void setLeaderAddress(String address) {
        log.debug("setLeaderAddress:{}",address);
        this.leaderAddress = address;
    }

    @Override
    public String getLeaderAddress() {
        return this.leaderAddress;
    }

    @Override
    public void actionHandlerRegistry(ActionHandlerRegistry actionHandlerRegistry, ExecutorService executor) {
        setActionHandlerRegistry(actionHandlerRegistry,executor);
    }

    @Override
    public void heartbeat() {
        RequestMessage<String> requestMessage = RequestMessageBuilder.builder()
                .body("ping")
                .action(RequestAction.HEARTBEAT)
                .build();
        try {
            oneWay(null,requestMessage,3_000);
        } catch (Exception e) {
            log.error(e.getMessage(),e);
        }
    }

    @Override
    public ResponseMessage sync(String addr, RequestMessage request, long timeout) throws Exception {
        Channel channel = getOrCreateChannel(addr);
        if(isChannelOk(channel)){
            try {
                return doSync(channel, request, timeout);
            }
            catch (TimeoutException e){
                return doSync(channel, request, timeout);
            }
        }
        else {
            closeChannel(channel);
        }
        return null;
    }

    @Override
    public void async(String addr, RequestMessage request, long timeout, Consumer<ResponseFuture> callback) throws Exception {
        Channel channel = getOrCreateChannel(addr);
        if(isChannelOk(channel)){
            try {
                doAsync(channel, request, timeout,callback);
            } catch (TimeoutException e) {
                doAsync(channel, request, timeout,callback);
            }
        }
        else {
            closeChannel(channel);
        }
    }

    @Override
    public void oneWay(String addr, RequestMessage request, long timeout) throws Exception {
        Channel channel = getOrCreateChannel(addr);
        if(isChannelOk(channel)){
            doOneWay(channel, request, timeout);
        }
        else {
            log.warn("channel not ok.");
            closeChannel(channel);
        }
    }

    private Channel getOrCreateChannel(String addr) throws InterruptedException {
        if(StringUtils.isBlank(addr)){
            addr = this.leaderAddress;
        }
        if(StringUtils.isBlank(addr)){
            log.warn("leader addr is blank.");
            throw new LeaderNotActiveException("leader addr is blank.");
        }
        NodeMetadata leader = NodeMetadataRepository.getInstance().getLeader();
        if(!leader.isActive()){
            log.error("{}",leader);
            BootstrapEventHolder.getInstance().publishEvent(new BootstrapEvent(1, BootstrapEventType.CHECK_LEADER));
            throw new LeaderNotActiveException("leader is not active.");
        }
        ChannelFuture channelFuture = channelFutureMap.get(addr);
        if(channelFuture != null){
            Channel channel = channelFuture.channel();
            if(isChannelOk(channel)) {
                return channel;
            }
        }
        //
        if(this.channelFutureMapLock.tryLock(this.nettyConfig.getConnectTimeoutMillis()*2L, TimeUnit.MILLISECONDS)){
            try {
                channelFuture = channelFutureMap.get(addr);
                boolean createNewChannel = true;
                if(channelFuture != null){
                    Channel channel = channelFuture.channel();
                    if(isChannelOk(channel)) {
                        return channel;
                    }
                    else if(!channelFuture.isDone()){
                        // 已存在未完成连接
                        createNewChannel = false;
                    }
                    else {
                        // 已失效
                        createNewChannel = true;
                        this.channelFutureMap.remove(addr);
                    }
                }
                else {
                    createNewChannel = true;
                }
                if(createNewChannel){
                    channelFuture = this.bootstrap.connect(AddressUtils.convertSocketAddress(addr));
                    log.info("reateChannel: begin to connect remote host {} asynchronously",addr);
                    this.channelFutureMap.put(addr,channelFuture);
                }
            } catch (Exception e) {
                log.error(e.getMessage(),e);
            }
            finally {
                this.channelFutureMapLock.unlock();
            }
        }
        else {
            log.error("try lock channelFutureMap timeout.");
        }
        try {
            if (channelFuture != null) {
                if (channelFuture.awaitUninterruptibly(this.nettyConfig.getConnectTimeoutMillis(), TimeUnit.MILLISECONDS)) {
                    Channel channel = channelFuture.channel();
                    if (isChannelOk(channel)) {
                        SocketAddress localAddress = channelFuture.channel().localAddress();
                        String local = AddressUtils.getStringSocketAddress(localAddress);
                        final int oldPort = DefaultNettyClient.this.port;
                        DefaultNettyClient.this.port = AddressUtils.parsePort(local);
                        log.info("createChannel: connect remote host[{}] success,{} -> {}", addr,oldPort, local);
                        createChannelErrorCount.set(0);
                        if(oldPort > 0){
                            // Leader变更或恢复，port变更
                            BootstrapEventHolder.getInstance().publishEvent(new BootstrapEvent(DefaultNettyClient.this.port, BootstrapEventType.CLIENT_CHANNEL_CONNECT));
                        }
                        else {
                            // 第一次连接
                            BootstrapEventHolder.getInstance().publishEvent(new BootstrapEvent(DefaultNettyClient.this.port, BootstrapEventType.CLIENT_STARTUP));
                        }
                        return channel;
                    } else {
                        log.error("createChannel: connect remote host[{}] failed, {}", addr, channelFuture.toString(), channelFuture.cause());
                        // 连续10次超时，检查状态，是否需要重启
                        if(createChannelErrorCount.incrementAndGet()>10){
                            BootstrapEventHolder.getInstance().publishEvent(new BootstrapEvent(1, BootstrapEventType.CHECK_LEADER));
                        }
                    }
                } else {
                    log.warn("createChannel: connect remote host[{}] timeout {}ms, {}", addr, this.nettyConfig.getConnectTimeoutMillis(), channelFuture.toString());
                    // 连续10次超时，检查状态，是否需要重启
                    if(createChannelErrorCount.incrementAndGet()>10){
                        BootstrapEventHolder.getInstance().publishEvent(new BootstrapEvent(1, BootstrapEventType.CHECK_LEADER));
                    }
                }
            }
        }
        finally {
            this.startCountDownLatch.countDown();
        }
        log.error("connect remote host[{}] failed",addr);
        return null;
    }



    class ChannelConnectManageHandler extends ChannelDuplexHandler {

        @Override
        public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress,
                            ChannelPromise promise) throws Exception {
            String local = AddressUtils.getStringSocketAddress(localAddress);
            String remote = AddressUtils.getStringSocketAddress(remoteAddress);

            log.info("netty client connect  {} => {}", local, remote);
            super.connect(ctx, remoteAddress, localAddress, promise);

        }

        @Override
        public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
            Channel channel = ctx.channel();
            String remote = AddressUtils.getChannelRemoteAddress(channel);
            log.info("netty client pipeline: disconnect {}", remote);

            super.disconnect(ctx, promise);

        }

        @Override
        public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
            Channel channel = ctx.channel();
            String remote = AddressUtils.getChannelRemoteAddress(channel);
            log.info("netty client pipeline: close {}", remote);

            super.close(ctx, promise);

        }

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            // 节点超时检测
            if(evt instanceof IdleStateEvent){
                IdleStateEvent event = (IdleStateEvent) evt;
                if(event.state().equals(IdleState.ALL_IDLE)){
                    Channel channel = ctx.channel();
                    final String remote = AddressUtils.getChannelRemoteAddress(channel);
                    log.warn("netty client pipeline: idle exception {}", remote);
                    BootstrapEventHolder.getInstance().publishEvent(new BootstrapEvent(DefaultNettyClient.this.port, BootstrapEventType.CLIENT_CHANNEL_IDLE));
                }
            }

            ctx.fireUserEventTriggered(evt);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            Channel channel = ctx.channel();
            final String remote = AddressUtils.getChannelRemoteAddress(channel);
            log.warn("netty client pipeline: exceptionCaught {}", remote);
            log.warn("netty client pipeline: exceptionCaught exception.", cause);
        }
    }
}
